package com.wjw.redis.stream.util;

import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 分布式延时队列工具类
 */
@Slf4j
@Component
@ConditionalOnBean({RedissonClient.class})
public class RedisDelayQueueUtil {

    @Resource
    private RedissonClient redissonClient;

    /**
     * 添加延迟队列
     *
     * @param value     队列值
     * @param delay     延迟时间
     * @param timeUnit  时间单位
     * @param queueCode 队列键
     * @param <T>
     */
    public <T> boolean addDelayQueue(@NonNull T value, long delay, @NonNull TimeUnit timeUnit, @NonNull String queueCode) {
        if (delay < 0) {
            log.error("(添加延时队列失败) 延迟时间不能为负数");
            return false;
        }
        if (StringUtils.isBlank(queueCode)) {
            return false;
        }
        if (!queueCode.matches("^[a-zA-Z0-9_-]+$")) {
            log.error("(添加延时队列失败) 队列键包含非法字符");
            return false;
        }
        try {
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            delayedQueue.offer(value, delay, timeUnit);
            //delayedQueue.destroy();
            log.info("(添加延时队列成功) 队列键：{}，队列值：{}，延迟时间：{}", queueCode, value, timeUnit.toSeconds(delay) + "秒");
        } catch (Exception e) {
            log.error("(添加延时队列失败) {}", e.getMessage());
            throw new RuntimeException("(添加延时队列失败)");
        }
        return true;
    }

    /**
     * 获取延迟队列
     *
     * @param queueCode 队列名称
     * @param <T> 队列值
     */
    public <T> T getDelayQueue(@NonNull String queueCode) {
        // 检查redissonClient是否已初始化
        if (redissonClient == null) {
            log.error("Redisson client is not initialized");
            return null;
        }

        if (StringUtils.isBlank(queueCode)) {
            log.error("Queue code is blank");
            return null;
        }

        RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
        // 这行代码用于触发redisson内的订阅服务
        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);

        try {
            // 使用poll代替take，避免阻塞，改为非阻塞方式
            Object value = blockingDeque.poll();
            if (value != null) {
                // 安全的类型转换，此处假设调用者明白返回值类型
                return (T) value;
            } else {
                // 队列为空
                log.debug("Queue is empty");
                return null;
            }
        } catch (Exception e) {
            // 捕获并记录可能的异常，避免中断调用方的执行
            log.error("Error when polling from queue: " + queueCode, e);
            return null;
        }
    }

    /**
     * 删除指定队列中的消息
     *
     * @param o         指定删除的消息对象队列值(同队列需保证唯一性)
     * @param queueCode 指定队列键
     */
    public boolean removeDelayedQueue(@NonNull Object o, @NonNull String queueCode) {
        if (StringUtils.isBlank(queueCode)) {
            return false;
        }

        try {
            // 获取BlockingDeque实例
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
            // 从blockingDeque中获取DelayedQueue实例
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);

            // 在删除前检查消息是否存在，提升删除操作的确定性
            if (!delayedQueue.contains(o)) {
                // 如果消息不存在于队列中，记录日志或做进一步处理
                log.info("尝试删除的消息 {} 不在队列 {} 中", o, queueCode);
                return false;
            }

            // 尝试删除指定消息
            return delayedQueue.remove(o);
        } catch (Exception e) {
            // 捕获并处理可能的异常，确保方法健壮性
            log.error("删除队列 {} 中的消息时发生异常：", queueCode, e);
            return false;
        }
    }
}
